package cn.common.annotation;


import cn.common.RetryService;
import cn.common.RetryWorkerService;
import cn.common.config.BasicRetryConfig;
import cn.common.config.Idc;
import cn.common.config.TsrSpringContextAware;
import cn.common.dto.DelayMessageDTO;
import cn.common.dto.RetryRequestParamDTO;
import cn.common.enums.RetryLevel;
import cn.common.exceptions.BasicRetryErrorCode;
import cn.common.exceptions.BasicRetryException;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import com.util.config.HttpResult;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.Signature;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.ExceptionClassifierRetryPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Copyright
 *
 * @Author WuHuang
 * @Date 2022/12/21
 */

@Aspect
@Component
public class BasicRetryAspect {
    private static final Logger log = LoggerFactory.getLogger(BasicRetryAspect.class);
    private ConcurrentHashMap<String, RetryTemplate> retryTemplateHolder = new ConcurrentHashMap();
    @Resource
    private KafkaTemplate<Integer, KafkaMetaMessage> kafkaRetryTemplate;
    @Resource
    private BasicRetryConfig basicRetryConfig;
    @Resource
    private RetryService retryService;

    public BasicRetryAspect() {
    }

    @Pointcut("@annotation(cn.common.annotation.BasicRetry)")
    public void BasicRetryAspect() {
    }

    @Around("BasicRetryAspect()")
    public Object around(ProceedingJoinPoint pjp) throws Throwable {
        if (!(pjp.getTarget() instanceof RetryWorkerService)) {
            throw new BasicRetryException(BasicRetryErrorCode.NO_IMPLEMENT_RETRY_STANDAND);
        } else {
            String workBeanName = pjp.getTarget().getClass().getName();
            if (!pjp.getSignature().getName().equals(RetryWorkerService.class.getMethods()[0].getName())) {
                throw new BasicRetryException(BasicRetryErrorCode.MUST_OVERIDER_STANDAND);
            } else {
                Object[] obj = pjp.getArgs();
                RetryRequestParamDTO retryRequestParamDTO = (RetryRequestParamDTO)obj[0];
                Method method = this.getMethod(pjp);
                String requestBodyBeanName = retryRequestParamDTO.getRequestBody().getClass().getName();
                BasicRetry basicRetry = (BasicRetry)method.getAnnotation(BasicRetry.class);

                boolean enableDelayRetry = basicRetry.enableDelayRetry();
                long retryIntervel = basicRetry.retryIntervel();
                int retryMaxCount = basicRetry.retryMaxCount();
                String businessName = basicRetry.businessName().getBusinessName();
                RetryLevel[] retryLevels = basicRetry.retryPolicys();

                List<RetryLevel> retryLevelList = Arrays.asList(retryLevels);
//                List<RetryLevel> sortedLevelList = (List)retryLevelList.stream().sorted(Comparator.comparing(RetryLevel::getLevel)).collect(Collectors.toList());
                List<RetryLevel> sortedLevelList = (List)retryLevelList.stream().sorted(Comparator.comparing(RetryLevel::getLevel)).collect(Collectors.toList());
                Object res = null;

                try {
                    res = pjp.proceed();
                    return res;
                } catch (BasicRetryException var27) {
                    Map<String, Object> metaMap = this.initMetaMap(retryRequestParamDTO.getMetaMap());
                    metaMap.put("basicRetry_workBeanName", workBeanName);
                    metaMap.put("basicRetry_requestBodyBeanName", requestBodyBeanName);
                    Integer curRetryCount = (Integer)metaMap.get("basicRetry_curRetryCount");
                    Boolean fromDelay = (Boolean)metaMap.get("basicRetry_fromRetry");
                    if (curRetryCount == 0 && curRetryCount < retryMaxCount && !fromDelay) {
                        RetryTemplate retryTemplate = this.productRetryTemplate(workBeanName, retryMaxCount, retryIntervel);
                        RetryWorkerService retryWorkerService = (RetryWorkerService)pjp.getThis();
                        if (enableDelayRetry) {
                            long initinalDelayTime = (long)retryMaxCount * retryIntervel / 1000L + (long)(retryMaxCount * 1);
                            long delayTime = initinalDelayTime + (long)((RetryLevel)sortedLevelList.get(0)).getSeconds();
                            this.submitDelayJob(workBeanName, retryRequestParamDTO, delayTime, sortedLevelList, 0, businessName);
                        }
                        return this.doRetry(retryTemplate, retryWorkerService, retryRequestParamDTO);
                    }else {

                    throw new BasicRetryException(var27.getMessage());
                    }

                }
            }
        }
    }

    private Map<String, Object> initMetaMap(Map<String, Object> metaMap) {
        if (Objects.isNull(metaMap.get("basicRetry_retryNO"))) {
            metaMap.put("basicRetry_retryNO", TsrSpringContextAware.getAppName() + "_" + Idc.getInstance().genenrator());
        }

        if (Objects.isNull(metaMap.get("basicRetry_fromRetry"))) {
            metaMap.put("basicRetry_fromRetry", Boolean.FALSE);
        }

        if (Objects.isNull(metaMap.get("basicRetry_curRetryCount"))) {
            metaMap.put("basicRetry_curRetryCount", 0);
        }

        return metaMap;
    }

    private RetryTemplate productRetryTemplate(String key, int retryMaxCount, long retryIntervel) {
        RetryTemplate retryTemplate = (RetryTemplate)this.retryTemplateHolder.get(key);
        if (Objects.nonNull(retryTemplate)) {
            return retryTemplate;
        } else {
            RetryTemplate template = new RetryTemplate();
            ExceptionClassifierRetryPolicy retryPolicy = new ExceptionClassifierRetryPolicy();
            Map<Class<? extends Throwable>, RetryPolicy> policyMap = Maps.newHashMap();
            policyMap.put(Exception.class, new SimpleRetryPolicy(retryMaxCount));
            retryPolicy.setPolicyMap(policyMap);
            FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
            backOffPolicy.setBackOffPeriod(retryIntervel);
            template.setRetryPolicy(retryPolicy);
            template.setBackOffPolicy(backOffPolicy);
            this.retryTemplateHolder.put(key, template);
            return template;
        }
    }

    private Method getMethod(ProceedingJoinPoint joinPoint) {
        Method method = null;

        try {
            Signature signature = joinPoint.getSignature();
            MethodSignature msig = null;
            if (!(signature instanceof MethodSignature)) {
                throw new IllegalArgumentException("该注解只能用于方法");
            }

            msig = (MethodSignature)signature;
            method = joinPoint.getTarget().getClass().getMethod(msig.getName(), msig.getParameterTypes());
        } catch (NoSuchMethodException var5) {
            log.error("annotation no sucheMehtod", var5);
        } catch (SecurityException var6) {
            log.error("annotation SecurityException", var6);
        }

        return method;
    }

    private HttpResult doRetry(RetryTemplate retryTemplate, RetryWorkerService retryWorkerService, RetryRequestParamDTO requetDto) {
        HttpResult result = null;

        try {
            result = this.retryService.retry(retryTemplate, requetDto, retryWorkerService);
            return result;
        } catch (Throwable var6) {
            log.error("doRetry", var6);
            return result;
        }
    }

    private void submitDelayJob(final String workBeanName, final RetryRequestParamDTO retryRequestParamDTO, final long initinalDelayTime, final List<RetryLevel> sortedLevelList, int currentSendMessageCount, final String businessName) {
        DelayMessageDTO delayMessageDTO = this.buildDelayMessageDTO(workBeanName, retryRequestParamDTO, initinalDelayTime, sortedLevelList, currentSendMessageCount, businessName);
        final KafkaMetaMessage kafkaMetaMessage = new KafkaMetaMessage();
        kafkaMetaMessage.setData(JSON.toJSONString(delayMessageDTO));
        final ListenableFuture<SendResult<Integer, KafkaMetaMessage>> result = this.kafkaRetryTemplate.send("basic_retry_submit_delay_job_topic_test", kafkaMetaMessage);
        result.addCallback(new ListenableFutureCallback<SendResult<Integer, KafkaMetaMessage>>() {
            public void onSuccess(SendResult<Integer, KafkaMetaMessage> resultx) {
                BasicRetryAspect.log.info("Retry Component send delay message success request={} ", JSON.toJSONString(resultx.getProducerRecord().value()));
            }

            public void onFailure(Throwable ex) {
                BasicRetryAspect.log.info("Retry Component send delay message fail ", JSON.toJSONString(kafkaMetaMessage));

                try {
                    KafkaMetaMessage retryKafkaMessage = (KafkaMetaMessage)((SendResult)result.get()).getProducerRecord().value();
                    DelayMessageDTO retryDelayMessageDTO = (DelayMessageDTO)JSON.parseObject(retryKafkaMessage.getData().toString(), DelayMessageDTO.class);
                    Integer sendMessageCount = retryDelayMessageDTO.getSendMessageCurrentCount();
                    if (sendMessageCount < BasicRetryAspect.this.basicRetryConfig.getSendMessageMaxCount()) {
                        BasicRetryAspect.this.submitDelayJob(workBeanName, retryRequestParamDTO, initinalDelayTime, sortedLevelList, sendMessageCount + 1, businessName);
                        return;
                    }

                    BasicRetryAspect.log.error("Retry Component send delay message fail maxRetryCount[{}]", BasicRetryAspect.this.basicRetryConfig.getSendMessageMaxCount());
                } catch (Exception var5) {
                    BasicRetryAspect.log.error("Retry Component retry send delay message fail", var5);
                }

            }
        });
    }

    private DelayMessageDTO buildDelayMessageDTO(String workBeanName, RetryRequestParamDTO retryRequestParamDTO, long initinalDelayTime, List<RetryLevel> sortedLevelList, int currentSendMessageCount, String businessName) {
        RetryRequestParamDTO target = new RetryRequestParamDTO();
        BeanUtils.copyProperties(retryRequestParamDTO, target);
        Map<String, Object> duplictMap = new HashMap();
        Map<String, Object> metaMap = retryRequestParamDTO.getMetaMap();
        duplictMap.put("basicRetry_retryNO", metaMap.get("basicRetry_retryNO"));
        duplictMap.put("basicRetry_requestBodyBeanName", metaMap.get("basicRetry_requestBodyBeanName"));
        duplictMap.put("basicRetry_workBeanName", metaMap.get("basicRetry_workBeanName"));
        duplictMap.put("basicRetry_curRetryCount", metaMap.get("basicRetry_curRetryCount"));
//        duplictMap.put("basicRetry_dataRouteTag", TsfTagUtils.getDataRouteTag());
        duplictMap.put("basicRetry_fromRetry", Boolean.TRUE);
        target.setMetaMap(duplictMap);
        DelayMessageDTO delayMessageDTO = DelayMessageDTO.builder().workBeanName(workBeanName).businessName(businessName)
                .body(JSON.toJSONString(target)).deayTime(initinalDelayTime).ttr(5).timeUnit(TimeUnit.SECONDS)
                .serviceName(TsrSpringContextAware.getAppName()).retryNo((String)target.getMetaMap()
                        .get("basicRetry_retryNO")).delayRetryMaxCount(sortedLevelList.size())
                .delayCurRetryCount(0).businessNo(workBeanName + "_" + "1").sortedLevelList(sortedLevelList)
                .sendMessageCurrentCount(currentSendMessageCount).build();
        return delayMessageDTO;
    }
}

